package com.shujia.stream

import org.apache.spark.streaming.dstream.ReceiverInputDStream
import org.apache.spark.streaming.{Duration, Durations, StreamingContext}
import org.apache.spark.{SparkConf, SparkContext}

object Demo1WordCOunt {
  def main(args: Array[String]): Unit = {


    val conf: SparkConf = new SparkConf()
      .setMaster("local[2]")
      .setAppName("wc")


    val sc: SparkContext = new SparkContext(conf)

    /**
      * 拆功能键spark streaming上下文对象
      *
      * Durations.seconds(5)  间隔多久计算一次（决定了计算的延迟时间）
      *
      */

    val ssc: StreamingContext = new StreamingContext(sc, Durations.seconds(1))


    /**
      * 连接socket获取数据
      *
      *
      * 在master上启动一个服务端
      *
      * yum install nc
      * nc -lk 9999
      *
      */

    val ds: ReceiverInputDStream[String] = ssc.socketTextStream("master", 9999)



    //打印结果
    ds
      .flatMap(_.split(","))
      .map((_, 1))
      .reduceByKey(_ + _)
      .print()


    //启动
    ssc.start()
    ssc.awaitTermination()

  }
}
